package dxn.nl.file.client.core;

import dxn.nl.file.client.model.dto.CsExportExcelHeadDto;
import com.alibaba.fastjson2.JSON;
import dxn.nl.file.client.model.dto.CsExportDealResult;
import dxn.nl.file.client.model.dto.CsExportDealTypeHeadDto;
import dxn.nl.file.client.model.dto.TransferMessageDto;
import dxn.nl.file.client.model.enums.ExportDealResultTypeEnum;
import dxn.nl.file.client.model.enums.ExportFileTypeEnum;
import dxn.nl.file.client.model.enums.ExportMessageTypeEnum;
import dxn.nl.file.client.model.enums.ExportQueryDataTypeEnum;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * excel导入导出处理handler
 */
public class ExportClientDealHandler extends SimpleChannelInboundHandler<TransferMessageDto> {
    private static final Logger log = LoggerFactory.getLogger(ExportClientDealHandler.class);
    //public static ScheduledExecutorService pool = Executors.newScheduledThreadPool(30);
    private final static EventExecutorGroup group = new DefaultEventExecutorGroup(20);

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        log.info(">>>>>>>>连接");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        SocketChannel socketChannel = (SocketChannel) ctx.channel();
        log.info(socketChannel.localAddress()+">>>>>>>>退出");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("",cause);
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
        if (evt instanceof IdleStateEvent event) {
            if (event.state().equals(IdleState.READER_IDLE)) {
                // System.out.println("长期没收到服务器推送数据");
                //可以选择重新连接
            } else if (event.state().equals(IdleState.WRITER_IDLE)) {
                // System.out.println("长期未向服务器发送数据");
                //发送心跳包
                TransferMessageDto csNioMessage = new TransferMessageDto();
                csNioMessage.setType(ExportMessageTypeEnum.heartbeat.code);
                ctx.writeAndFlush(csNioMessage);
            } else if (event.state().equals(IdleState.ALL_IDLE)) {
                System.out.println("ALL");
            }
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext context,TransferMessageDto messageDto) {
        log.info("当前线程：{}",Thread.currentThread().getName());
        group.submit(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                dealRead(context,messageDto);
                return null;
            }
        });

        log.info("go on：{}",Thread.currentThread().getName());
    }

    private void dealRead(ChannelHandlerContext context,TransferMessageDto messageDto)
    {
        try {
            SocketChannel socketChannel = (SocketChannel) context.channel();

            if(messageDto.getType() == ExportMessageTypeEnum.heartbeat.code)
            {
                // 心跳包，直接结束
                return ;
            }
            log.info("收到消息：{} {} {}",messageDto.getType(),messageDto.getConfig().getModeCode(),messageDto.getConfig().getId());

            messageDto.getConfig().setClient(ExportClientSocketManager.appSessionId);
            if(ExportMessageTypeEnum.deal_result.code == messageDto.getType())
            {
                // 如果处理结果返回的数据，直接给future赋值，触发结果
                ExportClientSocketManager.dealFutureData(messageDto);
                return ;
            }

            if(messageDto.getConfig().getFileType() == ExportFileTypeEnum.export.code)
            {
                // 处理导出
                dealExport(messageDto,socketChannel);
            }
            else
            {
                // 处理导入
                dealImport(messageDto,socketChannel);
            }
        }
        catch (Exception ex)
        {
            log.error("文件处理客户端处理异常：",ex);
        }
        finally {
            ReferenceCountUtil.release(messageDto);
        }
    }

    /**
     * 处理接受到的导出数据处理，处理逻辑
     * @param req req
     * @param channel channel
     */
    private void dealExport(TransferMessageDto req, Channel channel)
    {
        // 获取导出处理服务
        DealExportService reportService = CsExportManager.getExportDealService(req.getConfig().getModeCode());
        if (reportService == null) {
            sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,"应用没有对该模块做处理");
            log.warn(req.getConfig().getModeCode()+":应用没有对该模块做处理");
            return ;
        }

        try {
            if (req.getType() == ExportMessageTypeEnum.apply_deal.code) {
                checkCanDealData(reportService,req,channel);
                return;
            }
            if (req.getType() == ExportMessageTypeEnum.query_head.code)
            {
                // 获取表头信息
                CsExportDealTypeHeadDto headResult = reportService.getExcelHeads(req.getConfig(),req.getMessageData(),req.getData());
                if(headResult == null)
                {
                    sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,"获取表头结果为null");
                    log.warn(req.getConfig().getModeCode()+":获取表头结果为null");
                    return ;
                }

                // 转化结果
                TransferMessageDto resultPo = req.convertNewResultMessage(ExportMessageTypeEnum.deal_result.code
                        ,ExportDealResultTypeEnum.continue_deal.code);
                // 记录返回结果
                resultPo.setData(JSON.toJSONBytes(headResult));

                if(headResult.getExportType() == ExportQueryDataTypeEnum.stream)
                {
                    sendHeadAndDealStreamData(channel,resultPo,reportService);
//                    pool.execute(()->{ // 流式查询
//                        sendHeadAndDealStreamData(channel,resultPo,reportService);
//                    });
                    return;
                }
                else {
                    // 分页查询直接写入返回值即可
                    channel.writeAndFlush(resultPo);
                }
            }
            else if (req.getType() == ExportMessageTypeEnum.query_data.code)
            {
                CsExportDealResult dealResult = reportService.queryData(req.getConfig(),req.getMessageData(), req.getData());
                // 转化结果
                TransferMessageDto resultPo = req.convertNewResultMessage(ExportMessageTypeEnum.deal_result.code
                        ,dealResult.getDealResult().code,dealResult.getErrorMessage());
                resultPo.getMessageData().setCurrentParam(dealResult.getCurrentParam());

                if(dealResult.getData() != null)
                {
                    resultPo.setData(JSON.toJSONBytes(dealResult.getData()));
                }

                channel.writeAndFlush(resultPo);
            }
            else
            {
                sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,"code不正确");
            }
        }
        catch (Exception ex)
        {
            log.error("导出处理异常：",ex);
            sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,"处理异常"+ex.getMessage());
        }
    }

    private void dealImport(TransferMessageDto req,Channel channel)
    {
        DealImportService importService = CsExportManager.getUploadDealService(req.getConfig().getModeCode());

        if (importService == null) {
            sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,"应用没有对该模块做处理");
            return ;
        }

        try {
            if (req.getType() == ExportMessageTypeEnum.apply_deal.code) {
                // 申请导入
                checkCanDealData(importService,req,channel);
                return;
            }

            if (req.getType() == ExportMessageTypeEnum.query_head.code) {
                // 获取表头
                List<CsExportExcelHeadDto> headResult = importService.getExcelHeads(req.getConfig(),req.getMessageData(), req.getData());
                sendNoResultOk(channel,req,ExportMessageTypeEnum.deal_result.code,headResult);
                return;
            }

            if (req.getType() == ExportMessageTypeEnum.query_data.code) {
                String dealResult = importService.dealExcelData(req.getConfig(),req.getMessageData(), req.getData());
                if(StringUtils.hasText(dealResult))
                {
                    // 处理失败
                    sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,dealResult);
                }
                else {
                    // 处理成功
                    sendNoResultOk(channel,req,ExportMessageTypeEnum.deal_result.code,null);
                }
                return;
            }

            sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,"code不正确");
        }catch (Exception ex)
        {
            log.error("",ex);
            sendNoResultError(channel,req,ExportMessageTypeEnum.deal_result.code,ex.getMessage());
        }
    }

    /**
     * 流式导出，发送表头结果，等待结果并持续发送数据，直到数据发送完毕
     * @param channel channel
     * @param req req
     * @param reportService reportService
     */
    private void sendHeadAndDealStreamData(Channel channel, TransferMessageDto req, DealExportService reportService) {
        TransferMessageDto queryResultMessage = ExportClientSocketManager.sendDataToServer(req, channel, true);
        if (queryResultMessage.getConfig().getDealResult() != ExportDealResultTypeEnum.continue_deal.code) {
            // 不是继续处理，那就是服务端处理失败了，导出终止
            log.error(queryResultMessage.getConfig().getModeCode() + " 表头结果服务端处理失败:" + queryResultMessage.getConfig().getErrorMessage());
            return;
        }

        final TransferMessageDto messagePo = queryResultMessage.convertNewResultMessage();
        messagePo.setType(ExportMessageTypeEnum.send_data.code);

        String dealResult = null;

        try {
            dealResult = reportService.streamQueryData(req.getConfig(), req.getMessageData(), req.getData(), (sheetName, list) -> {
                messagePo.getConfig().setId(UUID.randomUUID().toString());
                messagePo.getMessageData().setSheetName(sheetName);
                if (CollectionUtils.isEmpty(list)) {
                    messagePo.setData(null);
                } else {
                    messagePo.setData(JSON.toJSONBytes(list));
                }

                TransferMessageDto resultMessage = ExportClientSocketManager.sendDataToServer(messagePo, channel, true);
                if (resultMessage.getConfig().getDealResult() == ExportDealResultTypeEnum.fail.code) {
                    String errorMessage = String.format("%s 服务端数据处理失败: %s", resultMessage.getConfig().getModeCode()
                            , resultMessage.getConfig().getErrorMessage());
                    log.error(errorMessage);
                    return errorMessage;
                }
                return null;
            });
        }
        catch (Exception ex)
        {
            log.error("",ex);
            dealResult = "客户端处理失败:"+ex.getMessage();
        }

        messagePo.getConfig().setId(UUID.randomUUID().toString());
        if(dealResult == null || dealResult.trim()=="")
        {
            messagePo.getConfig().setDealResult(ExportDealResultTypeEnum.end.code);
            messagePo.setData(null);
        }
        else
        {
            messagePo.getConfig().setDealResult(ExportDealResultTypeEnum.fail.code);
            messagePo.getConfig().setErrorMessage(dealResult);
            messagePo.setData(null);
        }

        ExportClientSocketManager.sendDataToServer(messagePo, channel, false);
    }

    /**
     * 判断 当前程序是否能够处理此次导出任务
     * @param service service
     * @param req req
     * @param channel channel
     */
    private void checkCanDealData(DealExportBaseService service,TransferMessageDto req,Channel channel)
    {
        boolean canDealData = service.canDealData(req.getConfig(),req.getMessageData());
        if (!canDealData)
        {
            // 如果不能处理，发送结束
            sendNoResultOk(channel,req,ExportMessageTypeEnum.deal_result.code
                    ,ExportDealResultTypeEnum.end.code,null);
        }
        else
        {
            // 如果可以处理，发送继续
            sendNoResultOk(channel,req,ExportMessageTypeEnum.deal_result.code
                    ,ExportDealResultTypeEnum.continue_deal.code,null);
        }
    }



    private void sendNoResultOk(Channel channel,TransferMessageDto req,int dealType,Object data)
    {
        sendNoResultOk(channel,req,dealType,ExportDealResultTypeEnum.continue_deal.code,data);
    }

    private void sendNoResultOk(Channel channel,TransferMessageDto req,int dealType,int dealResult,Object data)
    {
        TransferMessageDto resultPo = req.convertNewResultMessage(dealType
                ,dealResult);
        if(data != null)
        {
            byte[] bytes = JSON.toJSONBytes(data);
            resultPo.setData(bytes);
        }
        channel.writeAndFlush(resultPo);
    }

    private void sendNoResultError(Channel channel,TransferMessageDto req,int dealType,String message)
    {
        TransferMessageDto resultPo = req.convertNewResultMessage(dealType
                ,ExportDealResultTypeEnum.fail.code,"处理异常:"+message);
        channel.writeAndFlush(resultPo);
    }
}